package cn.doitedu.tests;

import com.alibaba.fastjson.JSON;
import lombok.*;
import org.apache.commons.lang3.RandomUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class DataMonitor {

    public static void main(String[] args) {

        for(int i=1;i<=10;i++) {
            new Thread(new GenDataRunnable(i)).start();
        }
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    @Getter
    @Setter
    public static class RecEvent {
        private long user_id;
        private String event_id;
        private long event_time;
        private String device_type;
        private String page_type;
        private Props properties;
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    @Getter
    @Setter
    public static class Props {
        private String rec_region_id;
        private String show_items;
        private String item_id;
        private String rec_tracking_id;
    }

    public static class GenDataRunnable implements Runnable {

        long user_id;
        KafkaProducer<String, String> producer;

        public GenDataRunnable(long user_id) {

            this.user_id = user_id;

            Properties kfkProperties = new Properties();
            kfkProperties.put("bootstrap.servers", "doitedu:9092");
            kfkProperties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            kfkProperties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            producer = new KafkaProducer<>(kfkProperties);


        }

        @Override
        public void run() {
            for (int i = 0; i < 100000; i++) {
                try {
                    // rec_show 事件
                    String rec_tracking_id = "rt_" + user_id + "_" + i;

                    Props props = new Props("recid_0"+RandomUtils.nextInt(1,5), "item1,item2,item3", null, rec_tracking_id);
                    RecEvent recEvent = new RecEvent(user_id, "rec_show", System.currentTimeMillis(), "mi6","page01" ,props);
                    //System.out.println(JSON.toJSONString(searchEvent));

                    producer.send(new ProducerRecord<>("dwd-events-detail",JSON.toJSONString(recEvent)));

                    Thread.sleep(1);

                    // rec_click事件
                    if (RandomUtils.nextInt(1, 10) % 3 == 0) {
                        for (int j = 0; j < RandomUtils.nextInt(1, 4); j++) {
                            props.setItem_id("item" + RandomUtils.nextInt(1, 4));
                            props.setShow_items(null);
                            RecEvent clickEvent = new RecEvent(user_id, "rec_click", System.currentTimeMillis(), "mi6", "page01",props);
                            //System.out.println(JSON.toJSONString(clickEvent));

                            producer.send(new ProducerRecord<>("dwd-events-detail",JSON.toJSONString(clickEvent)));
                            Thread.sleep(1);
                        }
                    }

                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }


            }

        }
    }

}
